// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: wsfuyibing <682805@qq.com>
// Date: 2024-12-05

package src

import (
	"context"
	"gitee.com/go-libs/log"
	"gitee.com/go-libs/process"
	"gitee.com/gomq/sr"
	amqp "github.com/rabbitmq/amqp091-go"
	"net/http"
	"sync"
)

type producer struct {
	conn    *amqp.Connection
	manager *manager
	mu      *sync.RWMutex
	name    string
	process process.Process
}

func (o *producer) Process() process.Process {
	return o.process
}

// Publish
// publishes a payload to rabbitmq server.
func (o *producer) Publish(ctx context.Context, msg *sr.Payload) (messageId string, err error) {
	var (
		channel        *amqp.Channel
		connection     *amqp.Connection
		exchange, _, _ = o.manager.client.GenerateExchange(0, msg.Topic)
		publishing     = o.manager.client.GeneratePublishingWithPayload(msg)
	)

	// Clean when produced.
	defer func() {
		if r := recover(); r != nil {
			log.Fatalfc(ctx, `[servlet=rabbitmq][producer] publishing caused an panic: %v`, r)

			if err == nil {
				err = sr.NewError(http.StatusInternalServerError, `%v`, r)
			}
		}
		o.manager.client.CloseChannel(ctx, channel)
	}()

	// Open connection.
	if connection, err = o.connect(ctx); err != nil {
		return
	}

	// Open channel.
	if channel, err = o.manager.client.OpenChannel(ctx, connection); err != nil {
		return
	}

	// Attach opentracing headers.
	for k, vs := range msg.Header {
		if len(vs) > 0 && vs[0] != "" {
			publishing.Headers[k] = vs[0]
		}
	}

	// Call publisher.
	if err = o.manager.client.Publish(ctx, channel, exchange, msg.Tag, publishing); err == nil {
		messageId = publishing.MessageId
	}
	return
}

// + Process

func (o *producer) Run(ctx context.Context) (err error) {
	for {
		select {
		case <-ctx.Done():
			return
		}
	}
}

// + Access

func (o *producer) connect(ctx context.Context) (conn *amqp.Connection, err error) {
	o.mu.Lock()

	// Check
	// previous connection.
	if conn = o.conn; conn != nil {
		// Reconnect
		// if previous connection is closed.
		if conn.IsClosed() {
			o.conn = nil
			o.mu.Unlock()
			return o.connect(ctx)
		}

		// Reuse
		// previous connection.
		o.mu.Unlock()
		return
	}

	// Create connection.
	if conn, err = o.manager.client.OpenConnection(ctx); err != nil {
		o.mu.Unlock()
		return
	}

	// Success connected.
	o.conn = conn
	o.mu.Unlock()
	return
}

func (o *producer) init() *producer {
	o.mu = new(sync.RWMutex)
	o.name = "rabbitmq-producer"
	o.process = process.New(o.name, o)
	return o
}
